package org.jpush.core;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.jpush.core.Channel.PacketInfo;
import org.jpush.packages.OutgoingPacket;
import org.jpush.utils.Pointer;

/**
 * A <code>AbstractChannelManager</code> will maintain all the Channel
 * as <code>ChannelInfo</code>, the <code>ChannelInfo</code> will be grouped
 * as <code>ChannelInfoGroup</code> by priority
 * 
 * @author Lei Wang
 *
 */
public abstract class AbstractChannelManager
{
    private Map<String, ChannelInfo> channelMap = 
    	new HashMap<String, ChannelInfo>();

    private List<ChannelInfoGroup> channelInfoGroupList;
    
    public AbstractChannelManager()
    {
        channelInfoGroupList = new LinkedList<ChannelInfoGroup>();
    }

    /**
     * Insert the <code>BroadcastQueue</code> into the 
     * <code>ChannelInfoGroup</code>, there is a list
     * of <code>ChannelInfoGroup</code> ordered by the 
     * priority, if there is a <code>ChannelInfoGroup</code>
     * with the priority provided, the <code>ChannelInfo</code>
     * will insert into the group directly, otherwise a new 
     * <code>ChannelInfoGroup</code> will be created first.
     * 
     * @param channel         The <code>Channel</code>
     * @param priority      The priority used to find the
     * 						<code>ChannelInfoGroup</code>
     * @param packetQuota   The quota of the packet
     * @return  The <code>ChannelInfo</code> created
     */
    private ChannelInfo insertChannelIntoOrderedChain(Channel channel, 
    		int priority, int packetQuota)
    {
        ChannelInfo channelInfo = new ChannelInfo(channel, packetQuota);

        //if list is empty, create one and insert channel there.
        if (channelInfoGroupList.isEmpty())
        {
        	ChannelInfoGroup channelInfoGroup = new ChannelInfoGroup(priority);
        	channelInfoGroup.insertChannelInfo(channelInfo);
        	channelInfoGroupList.add(channelInfoGroup);
            return channelInfo;
        }

        //Search for group in the list order by the priority
        ChannelInfoGroup after = null;

        int pos = 0;
        for (ChannelInfoGroup cif :  channelInfoGroupList)
        {
            if (priority >= cif.priority)
            {
            	after = cif;
                break;
            }
        }

        ChannelInfoGroup parentGroup = null;
        
        // If reaches the end, create a new group and append
        // into the queue
        if (after == null)
        {
            ChannelInfoGroup newGroup = new ChannelInfoGroup(priority);
            channelInfoGroupList.add(newGroup);

            parentGroup = newGroup;
        }
        // If a group with same priority found, insert directly
        else if (priority == after.priority)
        {
            parentGroup = after;
        }
        // Otherwise, insert a new group before the "after".
        else
        {
            ChannelInfoGroup newGroup = new ChannelInfoGroup(priority);
            channelInfoGroupList.add(pos, newGroup);

            parentGroup = newGroup;

        }

        // Insert into the group
        parentGroup.insertChannelInfo(channelInfo);
        return channelInfo;
    }

    /**
     * Search the list of <code>ChannelInfo</code> and
     * in each <code>ChannelInfo</code>, search the list
     * of packet and find the packet with the key provided
     * 
     * @param startAt     The start point of the ChannelInfo list
     * @param clientKey   The key of the packet
     * @param pPacketInfo The found PacketInfo
     * @param pFoundInfo  The found ChannelInfo
     * @return  the packet found
     */
    private static OutgoingPacket getPacketFromGroupChain(ChannelInfoGroup channelInfoGroup, int startAt , String clientKey, 
    		Pointer<PacketInfo> pPacketInfo, Pointer<ChannelInfo> pFoundInfo)
    {
        OutgoingPacket packet = null;
        ChannelInfo nextChannelInfo = null;

        for (int i = startAt; i < channelInfoGroup.itemList.size(); i ++)
        {
        	nextChannelInfo = channelInfoGroup.itemList.get(i);
            pFoundInfo.setValue(nextChannelInfo);

            packet = nextChannelInfo.channel.getNextPacket(clientKey, pPacketInfo);

            if(packet != null)
                return packet;

        }

        return null;
    }
    
    protected abstract void preEncodeOutgoingPacket(OutgoingPacket packet);
    protected abstract void deleteOutgoingPacket(OutgoingPacket packet);
    protected abstract void activateSubscribers(String channelName);

    protected void reportOnBeforePushPacket(String channelName) {}
    protected void reportOnAfterPacketIsSent(String channelName, String subscriberKey) {}

    /**
     * Create a new channel by the provided information
     * 
     * @param channelKey
     * @param maxPacket
     * @param requireSubscription
     * @param priority
     * @param packetQuota
     */
    public void createChannel(String channelKey, int maxPacket, boolean requireSubscription, int priority, int packetQuota)
    {
        Channel queue = new Channel(this, channelKey, requireSubscription);
        queue.setMaxPackets(maxPacket);

        ChannelInfo pPushInfo = insertChannelIntoOrderedChain(queue, priority, packetQuota);

        channelMap.put(channelKey, pPushInfo);
    }

    /**
     * Remove a channel by the key given
     * 
     * @param channelKey
     */
    public void RemoveChannel(String channelKey)
    {
    	channelMap.remove(channelKey);
    }

    /**
     * Subscribe a connection to a channel
     * 
     * @param connectionKey
     * @param channelKey
     * @return
     */
    public boolean subscribeConnection(String connectionKey, String channelKey)
    {
    	ChannelInfo info = channelMap.get(channelKey);
    	if (info == null)
    	{
    		return false;
    	}
    	else
    	{
    		info.channel.subscribeConnection(connectionKey);
    		return true;
    	}
    }

    public boolean unsubscribeConnection(String connectionKey, String channelKey)
    {
    	ChannelInfo info = channelMap.get(channelKey);
    	if (info == null)
    	{
    		return false;
    	}
    	else
    	{
    		info.channel.unsubscribeConnection(connectionKey);
    		return true;
    	}
    }

    /**
     * @see this{@link #pushPacket(OutgoingPacket, String, String, int)}
     * @param packet
     * @param channelName
     */
    public void pushPacket(OutgoingPacket packet, String channelName)
    {
    	pushPacket(packet, channelName, "", 0);
    }


    /**
     * @see this{@link #pushPacket(OutgoingPacket, String, String, int)}
     * @param packet
     * @param channelName
     * @param killKey
     */
    public void pushPacket(OutgoingPacket packet, String channelName, String killKey)
    {
    	pushPacket(packet, channelName, killKey, 0);
    }

    /**
     * @see push packet into a channel
     * @see {@link Channel#pushPacket(OutgoingPacket, String, int)}
     * 
     * @param packet
     * @param channelName
     * @param killKey
     */
    public void pushPacket(OutgoingPacket packet, String channelName, String killKey, int objectCategory)
    {
    	ChannelInfo info = channelMap.get(channelName);
    	if (info != null)
    	{
    		preEncodeOutgoingPacket(packet);
    		reportOnBeforePushPacket(channelName);
    		
    		//
    		info.channel.pushPacket(packet, killKey, objectCategory);
    		
    		activateSubscribers(channelName);
    	}
    }

    /**
     * Remove packet from a channel
     * @see {@link Channel#removePacket(String, int)}
     * 
     * @param killKey
     * @param objectCategory
     * @param channelKey
     */
    public void removePacket(String killKey, int objectCategory, String channelKey)
    {
    	ChannelInfo info = channelMap.get(channelKey);
    	if (info != null)
    	{
    		info.channel.removePacket(killKey, objectCategory);
    	}
    }

    /**
     * Remove Connection from all the channels
     * @see Channel#unsubscribeConnection(String)
     * @param clientKey
     */
    public void removeConnection(String clientKey)
    {
    	for (ChannelInfo info : channelMap.values())
    	{
    		info.channel.unsubscribeConnection(clientKey);
    	}
    }

    /**
     * dispose all packets of all channels
     * @see Channel#disposePacket()
     */
    public void disposeAllPackets()
    {
    	for (ChannelInfo info : channelMap.values())
    	{
    		info.channel.disposeAllPackets();
    	}
    }

    /**
     * Return xml format of text contains the names of channel
     * 
     * @return
     */
    public String getChannelsNames()
    {
    	StringBuffer sb = new StringBuffer();
    	for (String key : channelMap.keySet())
    	{
    		sb.append("<queue val=\"").append(key).append("\"/>");
    	}
    	
    	return sb.toString();
    }

    /**
     * 
     * @param subscriberKey
     * @param channelcastCtx
     * @param pPacketInfo
     * @param pFoundInfo
     * @return
     */
    public OutgoingPacket getNextPacket(String subscriberKey, ChannelContext channelcastCtx, Pointer<PacketInfo> pPacketInfo, Pointer<ChannelInfo> pFoundInfo)
    {
        OutgoingPacket packet = null;

        if (channelcastCtx.getChannelInfo() != null)
        {
            packet = getPacketFromGroupChain(channelcastCtx.getChannelInfo().parentGroup, channelcastCtx.getChannelInfo().parentGroup.itemList.indexOf(channelcastCtx.getChannelInfo()),
            		subscriberKey, pPacketInfo, pFoundInfo);

            if (packet != null)
                return packet;
        }

        //Nothing is found, so go up!
        for (ChannelInfoGroup group : channelInfoGroupList)
        {
            packet = getPacketFromGroupChain(group, 0, subscriberKey, pPacketInfo, pFoundInfo);

            if (packet != null)
                return packet;
        }

        return null;
    }

    public void disposePacket(PacketInfo packetInfo, String subscriberKey, ChannelInfo channelInfo, 
    		ChannelContext channelCtx, boolean blnSuccess)
    {
        reportOnAfterPacketIsSent(channelInfo.channel.getChannelName(), subscriberKey);

        channelInfo.channel.disposePacket(packetInfo, subscriberKey, blnSuccess);

        if (blnSuccess)
        {
            //QoS
            if (channelCtx.getChannelInfo() == channelInfo)
            {
                channelCtx.incrementSentPackets();
                if (channelCtx.getSentPackets() == channelInfo.packetQuota)
                {
                	List<ChannelInfo> channelInfoList = channelInfo.parentGroup.itemList;
                	int idx = channelInfoList.indexOf(channelInfo);
                	if (idx > 0 && idx < channelInfoList.size() - 1)
                	{
                        channelCtx.setChannelInfo(channelInfoList.get(idx + 1));
                	}
                	else
                	{
                        channelCtx.setChannelInfo(null);
                	}
                    channelCtx.setSentPackets(0);
                }
            }
            else
            {
                channelCtx.setChannelInfo(channelInfo);
                channelCtx.setSentPackets(1);
            }
        }
    }
    
    protected static class ChannelInfo
    {
    	public Channel channel;
    	public int packetQuota;
    	public ChannelInfoGroup parentGroup;

    	public ChannelInfo(Channel channel, int packetQuota)
    	{
    	    this.channel = channel;
    	    this.packetQuota = packetQuota;
    	}
    }

    /**
     * A <code>ChannelInfoGroup</code> will maintain a list of
     * <code>ChannelInfo</code> with same priority
     * 
     * @author Lei Wang
     *
     */
    protected static class ChannelInfoGroup
    {
    	public int priority;
    	public List<ChannelInfo> itemList;
    	
    	public ChannelInfoGroup(int uPriority)
    	{
    		this.priority = uPriority;
    		this.itemList = new LinkedList<ChannelInfo>();
    	}

    	public void insertChannelInfo(ChannelInfo channelInfo)
    	{
    	    channelInfo.parentGroup = this;

    	    // If empty list, add directly
    	    if (itemList.isEmpty())
    	    {
    	        itemList.add(channelInfo);
    	        return;
    	    }

    	    int pos = 0;
    	    
    	    // The list is order by packetQuota so the channelInfo
    	    // should be inserted in the proper position
    	    for(ChannelInfo ci : itemList)
    	    {
    	    	pos ++;
    	        if (channelInfo.packetQuota >= ci.packetQuota)
    	        {
    	            break;
    	        }
    	    }
    	    
    	    itemList.add(pos, channelInfo);
    	}
    }
    
    protected static class ChannelContext
    {
    	private int nSentPackets;
    	private ChannelInfo currentChannelInfo;
        
        public ChannelContext()
        {
        	this.nSentPackets = 0;
        	this.currentChannelInfo = null;
        }
        
        public ChannelInfo getChannelInfo()
        {
        	return this.currentChannelInfo;
        }
        
        public void setChannelInfo(ChannelInfo channelInfo)
        {
        	this.currentChannelInfo = channelInfo;
        }
        
        public int getSentPackets()
        {
        	return this.nSentPackets;
        }
        
        public void setSentPackets(int nSentPackets)
        {
        	this.nSentPackets = nSentPackets;
        }
        
        public void incrementSentPackets()
        {
        	this.nSentPackets = this.nSentPackets + 1;
        }
    }
}

